﻿using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Web.Http;
using System.Net.Http;
using PServiceBus.WebAPI.DTO;
using PServiceBus.WebAPI.Transports;
using System.Net;
using System.IO;
using System.Collections.Concurrent;
using System.Threading;
using PServiceBus.Services.Web;
using PServiceBus.Core.Runtime.Extensions;
using PServiceBus.Core.Manager;

namespace PServiceBus.WebAPI.Controllers {
    public partial class ESBController : ApiController {

        static ConcurrentDictionary<string, ConcurrentDictionary<string, BaseHttpTransport>> _streams =
            new ConcurrentDictionary<string, ConcurrentDictionary<string, BaseHttpTransport>>();

        static ConcurrentDictionary<string, ConcurrentDictionary<string, ManualResetEventSlim>> _pollings =
            new ConcurrentDictionary<string, ConcurrentDictionary<string, ManualResetEventSlim>>();

        const string INVALID_CONNECTION_MESSAGE = "invalid connection: could not start subscription";

        [AcceptVerbs(GET)]
        public HttpResponseMessage Stream([FromUri]StreamingInfo info) {

            if (info == null) throw new ArgumentNullException("info");
            if (info.Subscriber == null) throw new InvalidOperationException("info.Subscriber");
            if (info.TransportName == null) throw new InvalidOperationException("info.Transport");

            if (!_streams.ContainsKey(info.Subscriber))
                _streams[info.Subscriber] = new ConcurrentDictionary<string, BaseHttpTransport>();

            var transport = CurrentHttpTransport;

            if (transport == null) 
                return Request.CreateErrorResponse(HttpStatusCode.BadRequest,
                    "Could not resolve transport, be sure to include transport as querystring in request as [ serversent, httpstreaming, foreverframe ]");


            _streams[info.Subscriber][info.TransportName] = transport;

            var success = true;
            WebStreamManagementManager.Execute(webMgmt => success = webMgmt.Open(ClientToken, Headers));

            if (!success) {
                return Request.CreateErrorResponse(HttpStatusCode.Forbidden, INVALID_CONNECTION_MESSAGE);
            }

            CreateStreamingHandler(info);

            var response = Request.CreateResponse();

            response.Content = new PushStreamContent(OnStreamAvailable, transport.ContentType);
            response.Content.Headers.Add("TransportName", info.TransportName);
            response.Content.Headers.Add("Subscriber", info.Subscriber);
            return response;
        }

        private void CreateStreamingHandler(StreamingInfo info) {
            var subscriber = MessageBus.SelectSubscriber(info.Subscriber);
            info.Interval = Math.Max(info.Interval, 1);

            var httpTransport = _streams[info.Subscriber][info.TransportName];

            httpTransport.TransportName = info.TransportName;
            httpTransport.Subscriber = info.Subscriber;

            httpTransport.Handler = subscriber.OnMessageReceived<object>(info.TransportName,
                msg => {
                    var success = true;
                    WebStreamManagementManager.Execute(webMgmt => success = webMgmt.ShouldReceive(ClientToken, Headers));
                    if (!success) {
                        httpTransport.Close();
                        return;
                    }
                    httpTransport.Send(msg.ToJson());
                }, 
                error => {
                    if (httpTransport.IsConnected)
                        error.Continue = true;
                }, TimeSpan.FromMilliseconds(info.Interval), info.BatchSize);
        }


        internal static void EndStreaming(string subscriberName, string transportName = null) {
            if (String.IsNullOrWhiteSpace(subscriberName)) return;
            ConcurrentDictionary<string, BaseHttpTransport> transports;
            ConcurrentDictionary<string, ManualResetEventSlim> resets;
            BaseHttpTransport httpTransport;
            ManualResetEventSlim reset;
            if (_streams.TryGetValue(subscriberName, out transports)) {
                if (transportName != null) {
                    if (transports.TryGetValue(transportName, out httpTransport)) {
                        if (httpTransport.IsConnected) httpTransport.Close();
                        _streams[subscriberName].TryRemove(transportName, out httpTransport);
                    }
                } else {
                    foreach (var kv in transports) {
                        var transport = kv.Value;
                        if (transport.IsConnected) transport.Close();
                        _streams[subscriberName].TryRemove(kv.Key, out httpTransport);
                    }
                    _streams.TryRemove(subscriberName, out transports);
                }
            }
            if (_pollings.TryGetValue(subscriberName, out resets)) {
                if (transportName != null) {
                    if (resets.TryGetValue(transportName, out reset)) {
                        if (!reset.IsSet) reset.Set();
                        _pollings[subscriberName].TryRemove(transportName, out reset);
                    }
                } else {
                    foreach (var kv in resets) {
                        var rst = kv.Value;
                        if (!rst.IsSet) rst.Set();
                        _pollings[subscriberName].TryRemove(kv.Key, out reset);
                    }
                    _pollings.TryRemove(subscriberName, out resets);
                }
            }
        }


        private BaseHttpTransport CurrentHttpTransport {
            get {
                var transport = QueryString["transport"];
                var durable = Convert.ToBoolean(QueryString["durable"]);
                if (String.IsNullOrWhiteSpace(transport)) return null;
                var httpTransport = (transport == "serversent" ? new SSETransport() as BaseHttpTransport :
                    transport == "httpstreaming" ? new HttpStreamingTransport() as BaseHttpTransport :
                    transport == "foreverframe" ? new ForeverFrameTransport() as BaseHttpTransport
                    : null);
                httpTransport.Request = Request;
                httpTransport.IsDurable = durable;
                httpTransport.ClientToken = ClientToken;
                httpTransport.ConnectionInfo = new TransportConnectionInfo() { 
                    ConnectionID = ConnectionID,
                    UserName = UserName,
                    Password = Password
                };
                httpTransport.Headers = Headers;
                return httpTransport;
            }
        }


        private void OnStreamAvailable(Stream stream, HttpContent content, TransportContext context) {
            var subscriber = content.Headers.GetValues("Subscriber").First();
            var transport = content.Headers.GetValues("TransportName").First();

            _streams[subscriber][transport].Writer = new StreamWriter(stream) { AutoFlush = true };
        }


        private const int REQUEST_TIME_OUT = 30 * 1000;
        private const int LONG_POLLING_INTERVAL = 10;

        [AcceptVerbs(POST)]
        public IList<object> NextMessages(StreamingInfo info) {

            var success = true;
            WebStreamManagementManager.Execute(webMgmt => success = webMgmt.Open(ClientToken, Headers));

            if (!success) {
                throw new InvalidOperationException(INVALID_CONNECTION_MESSAGE);
            }
            
            var subscriber = MessageBus.SelectSubscriber(info.Subscriber);
            IList<object> list = new List<object>();
            var reset = new ManualResetEventSlim(false);

            if (!_pollings.ContainsKey(info.Subscriber))
                _pollings[info.Subscriber] = new ConcurrentDictionary<string, ManualResetEventSlim>();

            _pollings[info.Subscriber][info.TransportName] = reset;

            dynamic handler = subscriber.OnMessageReceived<object>(info.TransportName,
                actionList: msgs =>
                {
                    success = true;
                    WebStreamManagementManager.Execute(webMgmt => success = webMgmt.ShouldReceive(ClientToken, Headers));
                    if (!success) {
                        throw new InvalidOperationException(INVALID_CONNECTION_MESSAGE);
                    }
                    if (msgs != null) {
                        list = msgs;
                        if (!reset.IsSet)
                            reset.Set();
                    }
                }, errorAction: error => error.Continue = true,
                interval: TimeSpan.FromMilliseconds(LONG_POLLING_INTERVAL),
                batchSize: info.BatchSize);

            reset.Wait(REQUEST_TIME_OUT);
            handler.Stop();
            return list;
        }
    }
}
